package com.overlord.im.server.message.service;

import com.overlord.im.commons.constant.AckType;
import com.overlord.im.commons.constant.OverlordImConstants;
import com.overlord.im.commons.message.OverlordImMessage;
import com.overlord.im.commons.mq.MqConstants;
import com.overlord.im.server.message.entity.OfflineMessageEntity;
import com.overlord.im.server.message.entity.ServerMessageEntity;
import com.overlord.im.server.mapstruct.MessageMapper;
import com.overlord.im.server.message.entity.MessageAckEntity;
import com.overlord.im.server.mq.sender.RedirectMsgMqSender;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * @ClassName SendMessageService
 * @Author wuwei
 * @Description NONE
 * @Date 2020/6/2 18:31
 **/
@Service
public class SendMessageService {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private RedirectMsgMqSender redirectMsgMqSender;

    @Resource
    private ServerMessageService serverMessageService;

    @Resource
    private MessageAckService messageAckService;

    @Resource
    private OfflineMessageService offlineMessageService;

    @Resource
    private MessageMapper messageMapper;


    /**
     * 发送MSG_N消息至接收方
     *
     * @param message
     * @param entity
     */
    public void sendMsgN(OverlordImMessage message, ServerMessageEntity entity) {
        ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();
        //给目标发送MSG:N
        String hostPort = operations.get(message.getTo().toString());
        if (StringUtils.isNotBlank(hostPort)) {
            redirectMsgMqSender.send(toMsgN(message, entity), MqConstants.OVERLORD_IM_MESSAGE_DO_REDIRECT + hostPort,
                    MqConstants.OVERLORD_IM_MESSAGE_OVER_TIME_EXCHANGE);
            serverMessageService.saveOrUpdate(entity);
            return;
        }
        //如果没有找到节点的话说明是离线消息
        OfflineMessageEntity offlineMessageEntity = toOfflineMessage(entity);
        offlineMessageService.save(offlineMessageEntity);

    }

    /**
     * 通知发送方消息服务器已接收
     *
     * @param message
     * @param entity
     */
    public void sendMsgA(OverlordImMessage message, ServerMessageEntity entity) {
        ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();
        //给发送方发送MSG:A
        String senderHostPort = operations.get(message.getFrom().toString());
        if (StringUtils.isNotBlank(senderHostPort)) {
            MessageAckEntity ackEntity = messageMapper.toAck(entity);
            ackEntity.setAckType(AckType.MSG_A);
            messageAckService.save(ackEntity);
            redirectMsgMqSender
                    .send(toMsgA(message, entity), MqConstants.OVERLORD_IM_MESSAGE_DO_REDIRECT + senderHostPort,
                            MqConstants.OVERLORD_IM_MESSAGE_OVER_TIME_EXCHANGE);
            return;
        }
    }

    /**
     * 发送ACK_A消息回复接收方已确认接收消息
     *
     * @param message
     * @param entity
     */
    public void sendAckA(OverlordImMessage message, ServerMessageEntity entity) {
        ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();
        String senderHostPort = operations.get(entity.getMsgTo().toString());
        if (StringUtils.isNotBlank(senderHostPort)) {
            MessageAckEntity ackEntity = messageMapper.toAck(entity);
            ackEntity.setAckType(AckType.ACK_A);
            messageAckService.save(ackEntity);
            redirectMsgMqSender
                    .send(toAckA(message, entity), MqConstants.OVERLORD_IM_MESSAGE_DO_REDIRECT + senderHostPort,
                            MqConstants.OVERLORD_IM_MESSAGE_OVER_TIME_EXCHANGE);

            return;
        }
        //TODO 离线状态
    }

    /**
     * 通知发送方消息已送达
     *
     * @param message
     * @param entity
     */
    public void sendAckN(OverlordImMessage message, ServerMessageEntity entity) {
        ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();
        //给目标发送ACK:N
        String hostPort = operations.get(message.getFrom().toString());
        if (StringUtils.isNotBlank(hostPort)) {
            MessageAckEntity ackEntity = messageMapper.toAck(entity);
            ackEntity.setAckType(AckType.ACK_N);
            messageAckService.save(ackEntity);
            redirectMsgMqSender.send(toAckN(message, entity), MqConstants.OVERLORD_IM_MESSAGE_DO_REDIRECT + hostPort,
                    MqConstants.OVERLORD_IM_MESSAGE_OVER_TIME_EXCHANGE);
            return;
        }

        //TODO 离线状态
    }

    private OverlordImMessage toAckA(OverlordImMessage message, ServerMessageEntity entity) {
        return OverlordImMessage.builder().clientMessageId(message.getClientMessageId()).msgType(message.getMsgType())
                .ackType(AckType.ACK_A).sequenceId(entity.getSequenceId())
                .to(entity.getMsgTo()).version(OverlordImConstants.OVERLORD_IM_VERSION).build();
    }

    private OfflineMessageEntity toOfflineMessage(ServerMessageEntity entity) {
        return OfflineMessageEntity.builder().messageId(entity.getId()).sequenceId(entity.getSequenceId())
                .msgFrom(entity.getMsgFrom()).msgTo(entity.getMsgTo()).build();
    }

    private OverlordImMessage toAckN(OverlordImMessage message, ServerMessageEntity entity) {
        return OverlordImMessage.builder().clientMessageId(message.getClientMessageId()).msgType(message.getMsgType())
                .ackType(AckType.ACK_N).sequenceId(entity.getSequenceId())
                .to(message.getFrom()).version(OverlordImConstants.OVERLORD_IM_VERSION).build();
    }

    private OverlordImMessage toMsgN(OverlordImMessage message, ServerMessageEntity entity) {
        message.setAckType(AckType.MSG_N);
        message.setSequenceId(entity.getSequenceId());
        return message;
    }


    private OverlordImMessage toMsgA(OverlordImMessage message, ServerMessageEntity entity) {
        return OverlordImMessage.builder().clientMessageId(message.getClientMessageId()).msgType(message.getMsgType())
                .ackType(AckType.MSG_A).sequenceId(entity.getSequenceId())
                .to(message.getFrom()).version(OverlordImConstants.OVERLORD_IM_VERSION).build();
    }
}
